package com.navinfo.tripanalysis.offline.service.impl.huashen;

import com.navinfo.tripanalysis.common.pojo.PointProtocol;
import com.navinfo.tripanalysis.common.pojo.ProtocolEnum;
import com.navinfo.tripanalysis.common.pojo.RunningType;
import com.navinfo.tripanalysis.common.pojo.RunningTypeEnum;
import com.navinfo.tripanalysis.offline.service.PointProtocolConvertService;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.catalyst.expressions.GenericRow;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;

/**
 * 统一协议落盘数据类型转换接口实现类
 * @author hmc
 */
@RunningType(value = RunningTypeEnum.HS, interfaceClass = PointProtocolConvertService.class )
public class PointProtocolConvertServiceImpl implements PointProtocolConvertService {

    @Override
    public Row toRow(PointProtocol pointProtocol) {
        List<String> list = new ArrayList<>();
        if(ProtocolEnum.LOCATION.equals(pointProtocol.getProtocol())){
            list.add(String.valueOf(pointProtocol.getTid()));
            list.add(String.valueOf(pointProtocol.getTripId()));
            list.add("0200");
            list.add(String.valueOf(pointProtocol.getGpsTime()));
            list.add(String.valueOf(pointProtocol.getSpeedKmh()*100));
            list.add(pointProtocol.getOriginalLat()+"");
            list.add(pointProtocol.getOriginalLng()+"");
            list.add(pointProtocol.getLatitude()+"");
            list.add(pointProtocol.getLongitude()+"");
            list.add(pointProtocol.getDirection()+"");
            list.add(pointProtocol.getHeight()+"");
            list.add(pointProtocol.getEngineOutputTorque()*100+"");
            list.add(pointProtocol.getRotation()*100+"");
            list.add(pointProtocol.getAccPedalPos()*100+"");
            list.add(null);//0200没有制动信号
            list.add(pointProtocol.getClutchSwitch()*100+"");
            list.add(pointProtocol.getRealTimeOilConsumption()*100+"");
            list.add(pointProtocol.getFuelConsumptionRate()*100+"");
            list.add(null);//0200没有大气温度
            //0200特有
            list.add(pointProtocol.getAccStatus()+"");
            list.add(pointProtocol.getGpsMileage()==null ? null : pointProtocol.getGpsMileage()*1000+"");
            list.add(pointProtocol.getEcuMileage()==null ? null : pointProtocol.getEcuMileage()*100+"");
            list.add(pointProtocol.getMeterMileage()==null ? null :pointProtocol.getMeterMileage()*100+"");
            list.add(pointProtocol.getIntegralMileage()==null ? null: pointProtocol.getIntegralMileage()*100+"");
            list.add(pointProtocol.getStandardMileage()==null ? null : pointProtocol.getStandardMileage()+"");
            list.add(pointProtocol.getIntegralFuelCons()==null ? null :pointProtocol.getIntegralFuelCons()*100+"");
            list.add(pointProtocol.getTotalFuelCons()==null ? null : pointProtocol.getTotalFuelCons()*100+"");
            list.add(pointProtocol.getStandardFuelCons()==null ? null :pointProtocol.getStandardFuelCons()+"");
            list.add(pointProtocol.getCoolingWaterTem()==null ? null :pointProtocol.getCoolingWaterTem()*100+"");
            list.add(pointProtocol.getIntakeAirTem()==null ? null : pointProtocol.getIntakeAirTem()*100+"");
            list.add(pointProtocol.getIntakePressure()==null ? null :pointProtocol.getIntakePressure()*100+"");
            list.add(pointProtocol.getCumulativeRunningTime()*100+"");
            list.add(pointProtocol.getCruiseControl()+"");
            list.add(pointProtocol.getOilPercentage()==null ? null :pointProtocol.getOilPercentage()*100+"");
            list.add(pointProtocol.getHandBrakeStatus()+"");
            list.add(pointProtocol.getSlideNeutralStatus()+"");
            list.add(pointProtocol.getRapidAccelerationStatus()+"");
            list.add(pointProtocol.getRapidDecelerationStatus()+"");
            list.add(pointProtocol.getSharpTurningStatus()+"");
            list.add(pointProtocol.getSlamAcceleratorStatus()+"");
            list.add(null==pointProtocol.getAmtLoad() ? null : pointProtocol.getAmtLoad()+"");
            list.add(null==pointProtocol.getVecuLoad() ? null : pointProtocol.getVecuLoad()+"");
            list.add(pointProtocol.getPatch()+"");
            list.add(pointProtocol.getSerialNumber()+"");
            list.add(pointProtocol.getDataUpd());
            list.add(String.valueOf(pointProtocol.getTid()%32));
        }else if(ProtocolEnum.REALTIME.equals(pointProtocol.getProtocol())) {
            list.add(String.valueOf(pointProtocol.getTid()));
            list.add(String.valueOf(pointProtocol.getTripId()));
            list.add("0f37");
            list.add(String.valueOf(pointProtocol.getGpsTime()));
            list.add(String.valueOf(pointProtocol.getSpeedKmh() * 100));
            list.add(pointProtocol.getOriginalLat() + "");
            list.add(pointProtocol.getOriginalLng() + "");
            list.add(pointProtocol.getLatitude() + "");
            list.add(pointProtocol.getLongitude() + "");
            list.add(pointProtocol.getDirection() + "");
            list.add(pointProtocol.getHeight() + "");
            list.add(pointProtocol.getEngineOutputTorque() * 100 + "");
            list.add(pointProtocol.getRotation() * 100 + "");
            list.add(pointProtocol.getAccPedalPos() * 100 + "");
            list.add(pointProtocol.getBrake() * 100 + "");
            list.add(pointProtocol.getClutchSwitch() * 100 + "");
            list.add(pointProtocol.getRealTimeOilConsumption() * 100 + "");
            list.add(pointProtocol.getFuelConsumptionRate() * 100 + "");
            list.add(pointProtocol.getAtmosphericTem()==null ? null : pointProtocol.getAtmosphericTem() * 100 + "");
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(null);
            list.add(pointProtocol.getSerialNumber() + "");
            list.add(pointProtocol.getDataUpd());
            list.add(String.valueOf(pointProtocol.getTid()%32));
        }

        return new GenericRow(list.toArray());
    }

    @Override
    public PointProtocol fromRow(Row row) {
        PointProtocol point = new PointProtocol();

        //公共属性赋值
        point.setTid(Long.parseLong(row.getAs("tid")));
        point.setTripId(Integer.parseInt(row.getAs("tripId")));
        point.setProtocol("0200".equals(row.getAs("protocol")) ? ProtocolEnum.LOCATION: ProtocolEnum.REALTIME);
        point.setGpsTime(Long.parseLong(row.getAs("gpsTime")));
        point.setSpeedKmh(Double.parseDouble(row.getAs("speed")) / 100.0);
        point.setOriginalLat(Integer.parseInt(row.getAs("originalLat")));
        point.setOriginalLng(Integer.parseInt(row.getAs("originalLng")));
        point.setLatitude(Integer.parseInt(row.getAs("latitude")));
        point.setLongitude(Integer.parseInt(row.getAs("longitude")));
        point.setDirection(Integer.parseInt(row.getAs("direction")));
        point.setHeight(Double.parseDouble(row.getAs("height")));
        point.setEngineOutputTorque(Integer.parseInt(row.getAs("torque")) / 100);
        point.setRotation(Integer.parseInt(row.getAs("rotation")) / 100);
        point.setAccPedalPos(Integer.parseInt(row.getAs("accelerator")) / 100);
        Optional.ofNullable((String)row.getAs("brake")).ifPresent(e ->point.setBrake(Integer.parseInt(e) / 100 )  );
        point.setClutchSwitch(Integer.parseInt(row.getAs("clutchSwitch")) / 100);
        point.setRealTimeOilConsumption(Float.parseFloat(row.getAs("realtimeFuelConsumption")) / 100.0);
        point.setFuelConsumptionRate(Float.parseFloat(row.getAs("fuelConsumptionRate")) / 100.0);

        //0F37特有属性赋值
        Optional.ofNullable((String)row.getAs("atmosphericTem")).ifPresent(e -> point.setAtmosphericTem(Integer.parseInt(e)/ 100 ) );

        //0200特有属性赋值
        Optional.ofNullable((String)row.getAs("accStatus")).ifPresent(e ->point.setAccStatus(Integer.parseInt(e)));
        Optional.ofNullable((String)row.getAs("gpsMileage")).ifPresent(e ->point.setGpsMileage(new BigDecimal(Double.parseDouble(e)).divide( new BigDecimal("1000"), 0, BigDecimal.ROUND_HALF_UP).doubleValue()));
        Optional.ofNullable((String)row.getAs("ecuMileage")).ifPresent(e -> point.setEcuMileage(Double.parseDouble(e) / 100.0) );
        Optional.ofNullable((String)row.getAs("meterMileage")).ifPresent(e ->point.setMeterMileage(Double.parseDouble(e)   / 100.0) );
        Optional.ofNullable((String)row.getAs("integralMileage")).ifPresent(e -> point.setIntegralMileage(Double.parseDouble(e) / 100.0) );
        Optional.ofNullable((String)row.getAs("standardMileage")).ifPresent(e -> point.setStandardMileage(Float.parseFloat(e)) );
        Optional.ofNullable((String)row.getAs("integralFuelCons")).ifPresent(e -> point.setIntegralFuelCons(Double.parseDouble(e)  / 100.0) );
        Optional.ofNullable((String)row.getAs("totalFuelCons")).ifPresent(e -> point.setTotalFuelCons(Double.parseDouble(e) / 100.0) );
        Optional.ofNullable((String)row.getAs("standardFuelCons")).ifPresent(e -> point.setStandardFuelCons(Float.parseFloat(e)) );
        Optional.ofNullable((String)row.getAs("coolingWaterTem")).ifPresent(e -> point.setCoolingWaterTem(Integer.parseInt(e) / 100) );
        Optional.ofNullable((String)row.getAs("intakeAirTem")).ifPresent(e -> point.setIntakeAirTem(Integer.parseInt(e) / 100) );
        Optional.ofNullable((String)row.getAs("intakePressure")).ifPresent(e -> point.setIntakePressure(Integer.parseInt(e) / 100) );
        Optional.ofNullable((String)row.getAs("cumulativeRunningTime")).ifPresent(e -> point.setCumulativeRunningTime(Long.parseLong(e)) );
        Optional.ofNullable((String)row.getAs("cruiseControl")).ifPresent(e -> point.setCruiseControl(Integer.parseInt(e)) );
        Optional.ofNullable((String)row.getAs("oilPercentage")).ifPresent(e -> point.setOilPercentage(Double.parseDouble(e)  / 100.0) );
        Optional.ofNullable((String)row.getAs("handBrakeStatus")).ifPresent(e -> point.setHandBrakeStatus(Integer.parseInt(e) ));
        Optional.ofNullable((String)row.getAs("slideNeutralStatus")).ifPresent(e -> point.setSlideNeutralStatus(Integer.parseInt(e) ));
        Optional.ofNullable((String)row.getAs("rapidAccelerationStatus")).ifPresent(e -> point.setRapidAccelerationStatus(Integer.parseInt(e) ));
        Optional.ofNullable((String)row.getAs("rapidDecelerationStatus")).ifPresent(e -> point.setRapidDecelerationStatus(Integer.parseInt(e) ));
        Optional.ofNullable((String)row.getAs("sharpTurningStatus")).ifPresent(e -> point.setSharpTurningStatus(Integer.parseInt(e) ));
        Optional.ofNullable((String)row.getAs("slamAcceleratorStatus")).ifPresent(e -> point.setSlamAcceleratorStatus(Integer.parseInt(e) ));
        Optional.ofNullable((String)row.getAs("amtLoad")).ifPresent(e -> point.setAmtLoad(Integer.parseInt(e) ));
        Optional.ofNullable((String)row.getAs("vecuLoad")).ifPresent(e -> point.setVecuLoad(Integer.parseInt(e) ));
        //Optional.ofNullable((String)row.getAs("cumulativeRunningTime")).ifPresent(e -> point.setCumulativeRunningTime(Long.parseLong(e)));
        Optional.ofNullable((String)row.getAs("isPatch")).ifPresent(e -> point.setPatch(Integer.parseInt(e) ));
        point.setSerialNumber(Integer.parseInt(Optional.ofNullable((String) row.getAs("serialNumber")).orElse("0")) );
        Optional.ofNullable((String)row.getAs("dataUpd")).ifPresent(e -> point.setDataUpd(e));

        return point;
    }

    @Override
    public List<StructField> createStructTypeList() {
        List<StructField> fieldList = new ArrayList<>();
        fieldList.add(new StructField("tid", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("tripId", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("protocol", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("gpsTime", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("speed", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("originalLat", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("originalLng", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("latitude", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("longitude", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("direction", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("height", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("torque", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("rotation", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("accelerator", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("brake", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("clutchSwitch", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("realtimeFuelConsumption", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("fuelConsumptionRate", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("atmosphericTem", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("accStatus", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("gpsMileage", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("ecuMileage", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("meterMileage", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("integralMileage", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("standardMileage", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("integralFuelCons", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("totalFuelCons", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("standardFuelCons", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("coolingWaterTem", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("intakeAirTem", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("intakePressure", DataTypes.StringType, true, Metadata.empty()));
        //增加累计运行时长，车次统计使用
        fieldList.add(new StructField("cumulativeRunningTime",DataTypes.StringType,true,Metadata.empty()));
        fieldList.add(new StructField("cruiseControl", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("oilPercentage", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("handBrakeStatus", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("slideNeutralStatus", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("rapidAccelerationStatus", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("rapidDecelerationStatus", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("sharpTurningStatus", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("slamAcceleratorStatus", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("amtLoad", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("vecuLoad", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("isPatch", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("serialNumber", DataTypes.StringType, true, Metadata.empty()));
        fieldList.add(new StructField("dataUpd", DataTypes.StringType, true, Metadata.empty()));

//        fieldList.add(new StructField("hashtid",DataTypes.StringType,true,Metadata.empty()));
        //STRUCT_TYPE = new StructType(fieldList.toArray(new StructField[0]));

        //fieldList.forEach(e -> COLUMN_NAMES.add(e.name()));
        return fieldList;
    }

    @Override
    public List<String> getPartitionCols() {
        return Arrays.asList("part_time","hashtid");
    }
}
